#include <unistd.h>
#include <string>
#include <sstream>
#include <zmq.hpp>
#include "zmqMsg.pb.h"
#include <Pluma/Pluma.hpp>
#include "HalconCpp.h"
#include "HDevThread.h"
#include "halconProcessorInterface.hpp"
#include <map>
#include <memory>
#include <thread>

using namespace std;
using namespace HalconCpp;

class HalconServer {
public:
    HalconServer();
    void start();

    ~HalconServer(){
        m_halconProcessorMap.clear();
        m_pluma.unloadAll();
    }

private:
    std::map<uint,std::shared_ptr<HalconProcessorInterface> > m_halconProcessorMap;
    pluma::Pluma m_pluma;
    zmq::context_t m_context;
    zmq::socket_t m_routerSocket,m_dealerSocket;
    std::vector<std::thread> m_threadPool;

    void initZmq();
    void initHalconProcessors();
    void initThreads();

    void worker_routine();
    void publishUpdate();
    void processGetReqList();
    void processMsg(zmq::message_t& request, std::string &replyStr);
    void parseImage(const zmqMsg::Image& msg, HObject& image);
    void convertResult2MsgStr(HalconResult& result, std::string& resultStr);


};


HalconServer::HalconServer():m_context(3),
    m_routerSocket(m_context,ZMQ_ROUTER),
    m_dealerSocket(m_context,ZMQ_DEALER)
{
}

void HalconServer::start()
{
    initHalconProcessors();
    initThreads();
    initZmq();
}

void HalconServer::initZmq()
{

    m_routerSocket.bind ("tcp://*:5555");
    m_dealerSocket.bind ("inproc://workers");
    zmq::proxy((void*)m_routerSocket, (void*)m_dealerSocket, NULL);

}

void HalconServer::initHalconProcessors()
{
    // one thread for halcon operators for muli-thread client request
    SetSystem("thread_num", 1);

    m_pluma.acceptProviderType<HalconProcessorInterfaceProvider>();

    // Load libraries
    m_pluma.loadFromFolder("plugins");

    // Get warrior providers into a vector
    std::vector<HalconProcessorInterfaceProvider*> providers;
    m_pluma.getProviders(providers);

    // Create a interface from each provider
    std::vector<HalconProcessorInterfaceProvider*>::iterator it;
    for (it = providers.begin() ; it != providers.end() ; ++it){
        // Create a warrior
        HalconProcessorInterface* interface = (*it)->create();
        // Display it's description
        std::cout << "The function id: "<<interface->getId()<<" is registered and"<<std::endl;
        std::cout <<"The description: "<<interface->getDescription()<< std::endl;
        m_halconProcessorMap.emplace(interface->getId(),std::shared_ptr<HalconProcessorInterface>(interface));
    }
}

void HalconServer::initThreads()
{
    int threadNum = sysconf(_SC_NPROCESSORS_CONF);
    std::cout<<"Set up "<<threadNum<<" threads for service"<<endl;
    for(int i=0; i< threadNum;i++) {
        m_threadPool.emplace_back(&HalconServer::worker_routine,this);
    }

    std::cout<<"Set up a thread for get reqList service"<<endl;
    m_threadPool.emplace_back(&HalconServer::processGetReqList,this);

    m_threadPool.emplace_back(&HalconServer::publishUpdate,this);
}

void HalconServer::worker_routine()
{
    zmq::socket_t socket(m_context, ZMQ_REP);
    socket.connect ("inproc://workers");

    while (true) {
        //  Wait for next request from client
        zmq::message_t request;
        socket.recv(&request);

        std::string resultStr;
        processMsg(request,resultStr);
        uint size=resultStr.length();

        //  Send reply back to client
        zmq::message_t reply(size);
        memcpy(reply.data (), resultStr.c_str(),size);
        socket.send(reply);

    }
}

void HalconServer::publishUpdate()
{
    zmq::socket_t publisher(m_context, ZMQ_PUB);
    publisher.bind("tcp://*:5565");
    sleep(1);

    while (true) {
        zmq::message_t message(10);
        publisher.send(message);
        cout<<"pulish update msg"<<endl;
        sleep(300);
    }


}

void HalconServer::processGetReqList()
{
    zmq::socket_t socket(m_context, ZMQ_REP);
    socket.bind("tcp://*:5559");
    while (true) {
        //  Wait for next request from client
        zmq::message_t request;
        socket.recv(&request);

        cout<<"Received get request list"<<std::endl;

        std::string resultStr;
        zmqMsg::RequestList requestList;
        zmq::message_t requestListMsg;

        std::map<uint,std::shared_ptr<HalconProcessorInterface> >::iterator it;
        for(it=m_halconProcessorMap.begin();it!=m_halconProcessorMap.end();++it)
        {
            std::shared_ptr<HalconProcessorInterface> processorP = it->second;
            zmqMsg::RequestList::Request* requestP = requestList.add_list();
            requestP->set_requesttype(processorP->getId());
            requestP->set_requestdescription(processorP->getDescription());
        }

        requestList.SerializeToString(&resultStr);
        uint size=resultStr.length();

        //  Send reply back to client
        zmq::message_t reply(size);
        memcpy(reply.data (), resultStr.c_str(),size);
        socket.send(reply);

    }

}

void HalconServer::processMsg(zmq::message_t &request, std::string &replyStr)
{
    HalconResult result;

    try{

        zmqMsg::Image imageMsg;
        imageMsg.ParseFromArray(request.data(),request.size());

        uint reqType = imageMsg.reqtype();
        std::map<uint,std::unique_ptr<HalconProcessorInterface> >::iterator it;
        if(m_halconProcessorMap.find(reqType) == m_halconProcessorMap.end()) {
            std::cout<<"The function Id: "<<reqType<<" is not implemented"<<std::endl;
            std::ostringstream os;
            os<<"The function Id: "<<reqType<<" is not implemented";
            result.failureReason = os.str();
            result.ifFail = true;
            convertResult2MsgStr(result,replyStr);
            return;
        }

        HObject image;
        parseImage(imageMsg,image);


        auto processor = m_halconProcessorMap[reqType];
        processor->process(image,result);
        convertResult2MsgStr(result,replyStr);
    }

    catch(HException& e){
        result.ifFail = true;
        result.failureReason = e.ErrorMessage();
        std::cout<<"HException happended: "<<result.failureReason<<endl;
    }
    catch(std::exception& e) {
        result.ifFail = true;
        result.failureReason = e.what();
        std::cout<<"std::exception happend"<<result.failureReason<<endl;
    }


}

void HalconServer::parseImage(const zmqMsg::Image& msg, HObject& image)
{
    int cols = msg.width();
    int rows = msg.height();
    int channelNum = msg.channelnum();
    int size=cols*rows;
    if(channelNum>1)
    {
        const std::string& redBuffer =msg.buffer2();
        const std::string& greenBuffer =msg.buffer1();
        const std::string& blueBuffer =msg.buffer0();
        GenImage3(&image,"byte",cols,rows,
                  (Hlong)(redBuffer.c_str()),(Hlong)(greenBuffer.c_str()),(Hlong)(blueBuffer.c_str()));
    } else {
        const std::string& buffer =msg.buffer0();
        GenImage1(&image,"byte",cols,rows,(Hlong)(buffer.c_str()));
    }

}

void HalconServer::convertResult2MsgStr(HalconResult &result, std::string& resultStr)
{
    zmqMsg::CircleMark resultMsg;

    uint num= result.circleMarkVector.size();
    for(uint i=0; i<num; i++){
        zmqMsg::CircleMark::Circle* circleP=resultMsg.add_roicircle();
        CircleMark& mark = result.circleMarkVector[i];

        circleP->set_centerx(mark.centerX);
        circleP->set_centery(mark.centerY);
        circleP->set_radius(mark.radius);
        circleP->set_ifwrong(mark.ifWrong);
        circleP->set_note(mark.note);

    }
    resultMsg.set_iffail(result.ifFail);
    resultMsg.set_faileddescription(result.failureReason);

    resultMsg.SerializeToString(&resultStr);

}


int main ()
{
    HalconServer server;
    server.start();

    return 0;
}
